package com.cet.reactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class EventProcessorExample {

    public static void main(String[] args) throws InterruptedException {
        // 创建事件处理器
        EventProcessor processor = new EventProcessor();
        
        // 创建事件Flux
        Flux<String> eventFlux = Flux.create(sink -> {
            // 创建事件监听器实现
            EventListener listener = new EventListener() {
                @Override
                public void onEvent(Event event) {
                    System.out.println("[Listener] 接收到事件: " + event.getData());
                    sink.next(event.getData());
                }
                
                @Override
                public void onComplete() {
                    System.out.println("[Listener] 事件流结束");
                    sink.complete();
                }
                
                @Override
                public void onError(Throwable e) {
                    System.out.println("[Listener] 发生错误: " + e.getMessage());
                    sink.error(e);
                }
            };
            
            // 注册监听器
            System.out.println("[Flux] 注册监听器");
            processor.registerListener(listener);
            
            // 设置取消和清理回调
            sink.onCancel(() -> {
                System.out.println("[Flux] 取消订阅，注销监听器");
                processor.unregisterListener(listener);
            });
            
            sink.onDispose(() -> {
                System.out.println("[Flux] 流处理完成，注销监听器");
                processor.unregisterListener(listener);
            });
        });
        
        // 订阅Flux
        System.out.println("准备订阅事件流...");
        eventFlux.subscribe(
            data -> System.out.println("[订阅者] 接收到数据: " + data),
            error -> System.err.println("[订阅者] 错误: " + error),
            () -> System.out.println("[订阅者] 事件流已完成")
        );
        
        // 模拟事件产生
        processor.generateEvents();
        
        // 保持程序运行
        Thread.sleep(2000);
    }
}

// 事件类
class Event {
    private final String data;
    
    public Event(String data) {
        this.data = data;
    }
    
    public String getData() {
        return data;
    }
}

// 事件监听器接口
interface EventListener {
    void onEvent(Event event);
    void onComplete();
    void onError(Throwable e);
}

// 事件处理器
class EventProcessor {
    private final List<EventListener> listeners = new ArrayList<>();
    private final ExecutorService executor = Executors.newSingleThreadExecutor();
    
    public void registerListener(EventListener listener) {
        synchronized (listeners) {
            listeners.add(listener);
        }
    }
    
    public void unregisterListener(EventListener listener) {
        synchronized (listeners) {
            listeners.remove(listener);
        }
    }
    
    public void generateEvents() {
        executor.execute(() -> {
            try {
                // 模拟产生5个事件
                for (int i = 1; i <= 5; i++) {
                    Thread.sleep(300); // 模拟处理延迟
                    Event event = new Event("事件-" + i);
                    notifyListeners(event);
                }
                
                // 模拟完成
                completeListeners();
            } catch (Exception e) {
                errorListeners(e);
            }
        });
    }
    
    private void notifyListeners(Event event) {
        synchronized (listeners) {
            for (EventListener listener : new ArrayList<>(listeners)) {
                listener.onEvent(event);
            }
        }
    }
    
    private void completeListeners() {
        synchronized (listeners) {
            for (EventListener listener : new ArrayList<>(listeners)) {
                listener.onComplete();
            }
        }
    }
    
    private void errorListeners(Throwable e) {
        synchronized (listeners) {
            for (EventListener listener : new ArrayList<>(listeners)) {
                listener.onError(e);
            }
        }
    }
}